package order;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * 顺序消息消费者
 *
 *  线程:ConsumeMessageThread_2 , 消息:Car(carId=1006, desc=检查)
 *  线程:ConsumeMessageThread_1 , 消息:Car(carId=1007, desc=检查)
 *  线程:ConsumeMessageThread_3 , 消息:Car(carId=1008, desc=检查)
 *  线程:ConsumeMessageThread_1 , 消息:Car(carId=1007, desc=松手刹)
 *  线程:ConsumeMessageThread_2 , 消息:Car(carId=1006, desc=松手刹)
 *  线程:ConsumeMessageThread_1 , 消息:Car(carId=1007, desc=挂挡)
 *  线程:ConsumeMessageThread_3 , 消息:Car(carId=1008, desc=松手刹)
 *  线程:ConsumeMessageThread_1 , 消息:Car(carId=1007, desc=踩油门)
 *  线程:ConsumeMessageThread_2 , 消息:Car(carId=1006, desc=挂挡)
 *  线程:ConsumeMessageThread_3 , 消息:Car(carId=1008, desc=挂挡)
 *  线程:ConsumeMessageThread_2 , 消息:Car(carId=1006, desc=踩油门)
 *  线程:ConsumeMessageThread_3 , 消息:Car(carId=1008, desc=踩油门)
 * @author Administrator
 * @date 2022/03/31 14:35
 **/
@Slf4j
public class OrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("Order_Consumer");
        consumer.setNamesrvAddr("192.168.147.151:9876;192.168.147.152:9876");

//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // topic tag
        consumer.subscribe("Order-Topic","order");

        // 回调函数
        consumer.registerMessageListener(new MessageListenerOrderly() {

                 @Override
                 public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                     for (MessageExt msg : msgs) {
                         log.info("线程:{} , 消息:{}", Thread.currentThread().getName(), new String(msg.getBody()));
                     }
                     return ConsumeOrderlyStatus.SUCCESS;
                 }
             }
          );
        // 开始消费
        consumer.start();
        log.info("开始消费");
    }
}
